﻿using System;
using System.Collections.Generic;
using System.Text;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using MapReduce.Utils;
using System.Timers;

namespace MapReduce.Comm
{
    public class ChunkJob{
        public String SlaveAddress;
        public byte[][] Input;
        public DateTime DateSent;
        public int Count;
        public ChunkJob(String pSlaveAddress, byte[][] pInput, int pCount)
        {
            SlaveAddress = pSlaveAddress;
            Input = pInput;
            DateSent = DateTime.Now;
            Count = pCount;
        }
    }//ChunkJob

    public class JobManager<T>
    {
        private IJob<T> mJob = null;

        private String mServiceName;
        private uint mTotalJobsDecrementCounter = 0;
        private uint mTotalJobs = 0;
        private Dictionary<String, System.Timers.Timer> mTimers = null;
        private Object mSlaveBackLock = new Object();

        private ushort mPort;
        private DateTime mStartTime;

        public Callbacks.JobProgress ProgressCallback = null;
        public Callbacks.JobDone DoneCallback = null;
        public Callbacks.JobError ErrorCallback = null;

        private JobManagerUtil<T> util = null;

        public JobManager(String pServiceName, IJob<T> pJob, ushort pPort)
        {
            mServiceName = pServiceName; 
            mJob = pJob;
            mPort = pPort;

            util = new JobManagerUtil<T>(mServiceName, mJob, mPort);
            
        }



        //You must wonder why count?? I don't feel like explaining it right now.
        public void Process(IEnumerable<KeyValuePair<String, T>> pInput, uint pCount)
        {
            if (ErrorCallback == null) throw new MissingMethodException("ErrorCallback missing");

            mTotalJobs = pCount;
            mTotalJobsDecrementCounter = pCount;
            mStartTime = DateTime.Now;

            //Map jobs to slaves
            long jobsPerSlave = pCount / SlaveManager.Free.Count;

            List<KeyValuePair<String, T>> chunkInput = new List<KeyValuePair<string, T>>();

            foreach (KeyValuePair<String, T> inputPair in pInput)
            {
                chunkInput.Add(inputPair);
                if (chunkInput.Count >= jobsPerSlave)
                {
                    ChunkJob cip = new ChunkJob(SlaveManager.GetNext(), util.GetIpArray(chunkInput), chunkInput.Count);
                    SendJobReq(cip);

                    //reset chunk input list
                    chunkInput = new List<KeyValuePair<string, T>>();
                }

            }//foreach pInput

            //send the remaining jobs
            if (chunkInput.Count > 0)
            {
                ChunkJob cip = new ChunkJob(SlaveManager.GetNext(), util.GetIpArray(chunkInput), chunkInput.Count);
                SendJobReq(cip);
            }
         
        }//Process

        Dictionary<String, ChunkJob> mDictSentJobs = new Dictionary<string, ChunkJob>();

        private void SendJobReq(ChunkJob pChunkJob )
        {
            //String pSlaveAddress, int pChunkCount, byte[][] pArr2d
            //mDictSentJobs[pSlaveAddress] = new ChunkJob(pSlaveAddress, pArr2d, pChunkCount);
            mDictSentJobs[pChunkJob.SlaveAddress] = pChunkJob;

            JobRequest req = new JobRequest(pChunkJob.SlaveAddress, mPort);


            //Two problems can happen
            // 1] Send socket fails for this slave
            // 2] socket connection goes through, but slave is unable to complete the job.
            //In both the cases, add this slave in an error and try next slave.

            req.Send(mServiceName, pChunkJob.Input, delegate(byte pStatus, byte[][] pOp)
            {
                StopAllSlavesTimer();

                lock (mSlaveBackLock)
                {

                    if (pStatus == JobRequest.SUCCESS)
                    {
                        mDictSentJobs.Remove(pChunkJob.SlaveAddress);

                        SlaveManager.ReleaseSlave(pChunkJob.SlaveAddress);

                        IEnumerable<KeyValuePair<String, String>> tmpOutput = (IEnumerable<KeyValuePair<String, String>>)U.Deserialize(pOp[0]);
                        ProgressCallback(tmpOutput);

                        //As soon as the first job comes back, start a timer to track bad slave jobs
                        StartSlowSlavesTimer();

                    }
                    else
                    {
                        SlaveErroredResend(pChunkJob.SlaveAddress, U.Convert(pOp[0]).ToString());
                        return;
                        //ErrorCallback(errorMsg);
                    }

                    mTotalJobsDecrementCounter -= (uint)pChunkJob.Count;


                    if (mTotalJobsDecrementCounter == 0)
                    {
                        if (mDictSentJobs.Count != 0)
                        {
                            throw new Exception("Something is not right.");
                        }
                        DoneCallback(null);
                    }

                }//lock( mSlaveBackLock )

            });


        }//SendJobReq

        void SlaveErroredResend(String pSlaveAddress, String pErrorMsg)
        {
            SlaveManager.ErrorSlave(pSlaveAddress, pErrorMsg);

            ChunkJob chunk = mDictSentJobs[pSlaveAddress];
            mDictSentJobs.Remove(pSlaveAddress);

            chunk.SlaveAddress = SlaveManager.GetNext();
            //And re-execute this job.
            SendJobReq(chunk);
        }


        void StopAllSlavesTimer()
        {
            if (mTimers != null)
            {
                //Stop all the timers
                foreach (KeyValuePair<string, System.Timers.Timer> pair in mTimers)
                {
                    pair.Value.Stop();
                }
                mTimers = null;
            }

        }

        void StartSlowSlavesTimer()
        {
            StopAllSlavesTimer();

            //Now create timers for all the pending jobs
            mTimers = new Dictionary<string, System.Timers.Timer>();

            //The timer will be = time till now * 2
            TimeSpan timeSinceStart = DateTime.Now - mStartTime;
            double waitFor = timeSinceStart.TotalMilliseconds * 2;

            foreach (KeyValuePair<String, ChunkJob> pair in mDictSentJobs)
            {
                System.Timers.Timer t = GetSlaveTimer(pair.Key, waitFor);
                mTimers[pair.Key] = t;
                t.Start();
            }



        }//StartSlowSlavesTimer


        private System.Timers.Timer GetSlaveTimer(String pSlaveAddress, double pWaitFor)
        {
            System.Timers.Timer t = new System.Timers.Timer(pWaitFor);
            t.Interval = pWaitFor;
            // Hook up the Elapsed event for the timer.
            t.Elapsed += new ElapsedEventHandler(delegate(Object pO, ElapsedEventArgs pEea)
            {
                //This event will be called for each slave when it is late to come back.
                
                //1. First check the status of this Slave by requesting Ping.
                SlaveManager.PingSlave(pSlaveAddress, delegate(byte pStatus, byte[][] pOp)
                {
                    if (pStatus == JobRequest.SUCCESS)
                    {
                        //Still a Good slave that happened to get a big job.

                    }
                    else
                    {
                        SlaveErroredResend(pSlaveAddress, U.Convert(pOp[0]).ToString());
                        return;
                    }
                });
                
                //If Fails, 
            });

            return t;
        }//GetSlaveTimer




    }//JobManager
}
